package com.zdp.nio.reactor.multi;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

public class MultiEchoReactor {
    ServerSocketChannel serverSocket;
    AtomicInteger next = new AtomicInteger(0);
    //selectors集合,引入多个selector选择器
    Selector[] selectors = new Selector[2];
    //引入多个子反应器
    SubReactor[] subReactors = null;

    public MultiEchoReactor(Integer port) throws IOException {
        // 初始化选择器
        selectors[0] = Selector.open();
        selectors[1] = Selector.open();
        // 初始化服务端
        serverSocket = ServerSocketChannel.open();
        serverSocket.configureBlocking(false);
        serverSocket.bind(new InetSocketAddress(port));
        // 注册选择器
        SelectionKey register = serverSocket.register(selectors[0], SelectionKey.OP_ACCEPT);
        register.attach(new AcceptorHandler());
        // 创建子反应器
        SubReactor subReactor1 = new SubReactor(selectors[0]);
        SubReactor subReactor2 = new SubReactor(selectors[1]);
        subReactors = new SubReactor[]{subReactor1,subReactor2};
    }

    public void startService(){
        new Thread(subReactors[0]).start();
        new Thread(subReactors[1]).start();
    }

    /**
     * 子反应器，一个子反应器对应一个Selector
     * 子反应器负责监听事件并分发
     */
    class SubReactor implements Runnable {

        final Selector selector;

        public SubReactor(Selector selector){
            this.selector = selector;
        }

        @Override
        public void run() {
            try{
                while(!Thread.interrupted()){
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()){
                        SelectionKey next = iterator.next();
                        dispatch(next);
                    }
                    selectionKeys.clear();
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        /**
         * 取出 事件绑定的处理器，调用该处理器进行事件处理
         * @param next
         */
        private void dispatch(SelectionKey next){
            Runnable handler = (Runnable) next.attachment();
            if(Objects.nonNull(handler)){
                handler.run();
            }
        }
    }

    /**
     * 监听 accpet事件，若有新连接进入，就将该连接注册到selector中
     * 同时为其的IO事件设置一个Handler处理器
     */
    class AcceptorHandler implements Runnable {
        public void run() {
            try {
                SocketChannel channel = serverSocket.accept();
                if (channel != null){
                    // 不同的连接设置到不同的连接器中 做取模运算
                    new MultiThreadEchoHandler(selectors[next.get()], channel);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            if (next.incrementAndGet() == selectors.length) {
                next.set(0);
            }
        }
    }

    public static void main(String[] args) throws IOException {
        MultiEchoReactor server =
                new MultiEchoReactor(9905);
        server.startService();
    }
}
